package core

import (
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"goetl/utils"

	"github.com/emirpasic/gods/lists/arraylist"
	"github.com/emirpasic/gods/sets/hashset"
)

const THREAD_COUNT string = "thread.count"

const UNIT_OF_WORK_INPUT_MESSAGE string = "Input Message"

const UNIT_OF_WORK_FLOW string = "Flow"

//步骤运行时线程
type StepRuntime struct {
	//是否运行中
	running bool
	//是否取消中
	cancelling bool
	//是否已取消
	cancelled bool

	//是否已结束
	finished bool

	//是否开始步骤
	StartStep bool
	//上下文
	context ComponentContext

	flowRuntime *FlowRuntime

	TargetStepRuntimes *arraylist.List
	SourceStepRuntimes *arraylist.List

	queueCapacity int32
	inQueue       *utils.Queue
	//读写锁
	sync.RWMutex

	activeCount int32

	ContentMessagesSentCount int32

	ControlMessagesSentCount int32

	liveSourceStepIds               *hashset.Set
	targetStepRuntimeUnitOfWorkSent *hashset.Set

	sourceStepRuntimeUnitOfWorkReceived map[string]bool
}

func NewStepRuntime(context ComponentContext, flowRuntime *FlowRuntime) *StepRuntime {
	queueCapacity := context.flowStep.ComponentId.GetInt("inbound.queue.capacity", 1000)
	inQueue := utils.NewQueue(int(queueCapacity))
	return &StepRuntime{context: context, running: false, cancelling: false, cancelled: false, finished: false, queueCapacity: queueCapacity, inQueue: inQueue, activeCount: 0, targetStepRuntimeUnitOfWorkSent: hashset.New(), sourceStepRuntimeUnitOfWorkReceived: make(map[string]bool), ContentMessagesSentCount: 0, ControlMessagesSentCount: 0}
}

func (this *StepRuntime) Start() {

}
func (this *StepRuntime) StartRunning() {
	this.running = true
}

//启动流程
func (this *StepRuntime) Run() {
	for this.running && !this.cancelled {

		var inputMessage *Message

		this.RLock()
		m, err := this.inQueue.GetNoWait() //队列取出消息

		if err == nil && m != nil {
			inputMessage = m.(*Message)

			if inputMessage.Type != "ShutdownMessage" {
				this.activeCount++
			}
		}
		this.RUnlock()

		if err == nil && m != nil {

			this.process(inputMessage)
		} else {

			time.Sleep(500 * time.Millisecond)
		}

	}
}

//加入队列
func (this *StepRuntime) Queue(message *Message) (err error) {
	if this.inQueue.Size() == int(this.queueCapacity) && (*message).Header.OriginatingStepId == this.context.flowStep.Id {
		err = errors.New("Inbound queue capacity on " + this.context.flowStep.ComponentId.Name + " not sufficient to handle inbound messages from other components in addition to inbound messages from itself.")

		return
	} else {
		if this.running {
			for qerr := this.inQueue.Put(message, 500); qerr == nil; {
				if this.cancelling {
					err = errors.New("flow is cancel")
					return
				}
			}
		}
	}

	return
}

//处理队列消息
func (this *StepRuntime) process(inputMessage *Message) {
	unitOfWorkBoundaryReached := this.calculateUnitOfWorkLastMessage(*inputMessage)
	fmt.Println(inputMessage.Type)

	this.processOnAnotherThread(inputMessage, unitOfWorkBoundaryReached)
}

func (this *StepRuntime) processOnAnotherThread(message *Message, unitOfWorkBoundaryReached bool) {

	component := this.context.flowStep.ComponentId

	fmt.Printf("log_input=%s\n", strconv.FormatBool(component.GetBoolean("log_input", false)))
	autoSendControlMessagesStr := ""
	if component.ComponentDefinition["autoSendControlMessages"] != nil {
		autoSendControlMessagesStr = component.ComponentDefinition["autoSendControlMessages"].(string)
	}
	if autoSendControlMessagesStr == "" {
		autoSendControlMessagesStr = "true"
	}
	fmt.Printf("autoSendControlMessages=%s\n", autoSendControlMessagesStr)
	autoSendControlMessages, _ := strconv.ParseBool(autoSendControlMessagesStr)
	fmt.Println(autoSendControlMessages)
	recursionDone := this.liveSourceStepIds.Size() == 1 && this.liveSourceStepIds.Contains(this.context.flowStep.Id) && this.getActiveCountPlusQueueSize() == 1
	fmt.Println(this.liveSourceStepIds.Size())

	if (unitOfWorkBoundaryReached || recursionDone) && autoSendControlMessages {
		this.verifyAndSendControlMessageToTargets(message)
	}

	if this.StartStep || recursionDone {
		this.shutdown()
	}
	this.decrementActiveCount()

}

func (this *StepRuntime) SetSourceStepRuntimes(sourceStepRuntimes *arraylist.List) {

	this.SourceStepRuntimes = sourceStepRuntimes

	this.liveSourceStepIds = hashset.New()

	this.SourceStepRuntimes.Each(func(i int, s interface{}) {
		stepRuntime := s.(*StepRuntime)
		fmt.Println((*stepRuntime).context.flowStep.Id)
		this.liveSourceStepIds.Add((*stepRuntime).context.flowStep.Id)

	})

	this.StartStep = sourceStepRuntimes.Empty()
}

func (this *StepRuntime) decrementActiveCount() {

	this.RLock()
	this.activeCount--
	this.RUnlock()
}
func (this *StepRuntime) getActiveCountPlusQueueSize() int32 {
	fmt.Println(this.activeCount + int32(this.inQueue.Size()))
	return this.activeCount + int32(this.inQueue.Size())
}
func (this *StepRuntime) shutdown() {

	this.finished = true
	this.running = false
	if this.cancelling {
		this.cancelled = true
	}

}

func (this *StepRuntime) calculateUnitOfWorkLastMessage(inputMessage Message) bool {
	lastMessage := true
	if inputMessage.Type == "ControlMessage" {
		if inputMessage.Header != nil {
			this.sourceStepRuntimeUnitOfWorkReceived[inputMessage.Header.OriginatingStepId] = true
		}
	}
	activeRuntimes := hashset.New()
	this.SourceStepRuntimes.Each(func(i int, s interface{}) {
		sourceRuntime := s.(*StepRuntime)

		var controlMessageReceived bool
		if _, ok := this.sourceStepRuntimeUnitOfWorkReceived[sourceRuntime.context.flowStep.Id]; ok {
			controlMessageReceived = true
		} else {
			controlMessageReceived = false
		}

		lastMessage = lastMessage && controlMessageReceived
		if !controlMessageReceived {
			activeRuntimes.Add(sourceRuntime)
		}

	})

	if !lastMessage {
		lastMessage = this.areStepRuntimesDead(activeRuntimes)
	}

	return lastMessage
}

func (this *StepRuntime) isQueueEmpty() bool {
	m, _ := this.inQueue.Peek(-1)
	return this.inQueue.IsEmpty() || m.(Message).Type == "ShutdownMessage"
}

func (this *StepRuntime) idle() bool {
	var r bool
	this.RLock()
	r = this.activeCount <= 0
	this.RUnlock()

	return r

}

func (this *StepRuntime) isStepRuntimeDead(stepRuntime StepRuntime) bool {
	var r = true
	if stepRuntime.ContentMessagesSentCount > 0 || stepRuntime.ControlMessagesSentCount > 0 || !stepRuntime.isQueueEmpty() || !stepRuntime.idle() {
		r = false
	} else {
		parentSteps := stepRuntime.SourceStepRuntimes
		parentSteps.Each(func(i int, s interface{}) {
			parentStep := s.(StepRuntime)
			parentStepIsDead := parentStep.isStepRuntimeDead(parentStep)
			if !parentStepIsDead && parentStep.ControlMessagesSentCount == 0 {

				r = false

			}
		})
	}
	return r
}

func (this *StepRuntime) areStepRuntimesDead(runtimes *hashset.Set) bool {
	for _, s := range runtimes.Values() {
		stepRuntime := s.(StepRuntime)
		if !this.isStepRuntimeDead(stepRuntime) {
			return false
		}
	}
	return true
}

func (this *StepRuntime) verifyAndSendControlMessageToTargets(inputMessage *Message) {
	fmt.Println("verifyAndSendControlMessageToTargets")
}
